package com.spring.controller;

import com.nebula.core.flink.NebulaEdgeSource;
import com.nebula.core.flink.NebulaVertexSource;
import com.nebula.core.util.NebulaFormat;
import com.spring.dao.PathDao;
import com.spring.dao.PlayerDao;
import com.spring.entity.GraphDo;
import com.spring.service.NebulaService;
import com.vesoft.nebula.client.graph.data.DateTimeWrapper;
import com.vesoft.nebula.client.graph.data.DateWrapper;
import com.vesoft.nebula.client.graph.data.TimeWrapper;
import com.vesoft.nebula.client.graph.data.ValueWrapper;
import com.vesoft.nebula.client.meta.MetaClient;
import com.vesoft.nebula.client.storage.StorageClient;
import com.vesoft.nebula.client.storage.data.BaseTableRow;
import com.vesoft.nebula.meta.ColumnDef;
import com.vesoft.nebula.meta.EdgeItem;
import com.vesoft.nebula.meta.TagItem;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.connector.nebula.connection.NebulaClientOptions;
import org.apache.flink.connector.nebula.connection.NebulaGraphConnectionProvider;
import org.apache.flink.connector.nebula.connection.NebulaMetaConnectionProvider;
import org.apache.flink.connector.nebula.connection.NebulaStorageConnectionProvider;
import org.apache.flink.connector.nebula.sink.NebulaEdgeBatchOutputFormat;
import org.apache.flink.connector.nebula.sink.NebulaVertexBatchOutputFormat;
import org.apache.flink.connector.nebula.statement.EdgeExecutionOptions;
import org.apache.flink.connector.nebula.statement.ExecutionOptions;
import org.apache.flink.connector.nebula.statement.VertexExecutionOptions;
import org.apache.flink.types.Row;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * 节点探索
 *
 * @author zhangtoa
 */
@RestController
@Slf4j
@RequestMapping("/test")
public class NebulaController {

    @Autowired
    private NebulaService nebulaService;

    @Resource
    private PathDao pathDao;

    @Resource
    private PlayerDao playerDao;

    @GetMapping("/findPlayerRelation")
    public Map<String, Object> findPlayerRelation(String vid, String edgeType) throws UnsupportedEncodingException {
        return NebulaFormat.formatResult(nebulaService.findPlayerRelation(vid, edgeType));
    }


    @GetMapping("/getPlayerProp")
    public Map<String, Object> getPlayerProp(String vid) throws UnsupportedEncodingException {
        return NebulaFormat.formatResult(nebulaService.getPlayerProp(vid));
    }

    @GetMapping("/findPath")
    public Map<String, Object> findPath(String src, String dst) throws UnsupportedEncodingException {
        return NebulaFormat.formatResult(nebulaService.findPath(src, dst));
    }

    @GetMapping("/getSubgraph")
    public Map<String, Object> getSubgraph(String vid, int steps) throws UnsupportedEncodingException {
        return NebulaFormat.formatResult(nebulaService.getSubgraph(vid, steps));
    }

    @GetMapping("/findDirectRelation")
    public Map<String, Object> findDirectRelation(String vid) throws UnsupportedEncodingException {
        return NebulaFormat.formatResult(nebulaService.findDirectRelation(vid));
    }

    @GetMapping("/test")
    public Object test() throws UnsupportedEncodingException {
        List<GraphDo> list = pathDao.selectBatch();
        return list;
    }

    @GetMapping("/scan")
    public Object scan() throws Exception {
        NebulaClientOptions nebulaClientOptions =
                new NebulaClientOptions.NebulaClientOptionsBuilder()
                        .setMetaAddress("10.218.221.239:9559")
                        .setGraphAddress("10.218.221.239:9669")
                        .build();
        NebulaStorageConnectionProvider storageConnectionProvider =
                new NebulaStorageConnectionProvider(nebulaClientOptions);

        NebulaMetaConnectionProvider metaConnectionProvider =
                new NebulaMetaConnectionProvider(nebulaClientOptions);

        NebulaGraphConnectionProvider graphConnectionProvider =
                new NebulaGraphConnectionProvider(nebulaClientOptions);

//        syncTag(storageConnectionProvider, metaConnectionProvider, graphConnectionProvider);
        syncEdge(storageConnectionProvider, metaConnectionProvider, graphConnectionProvider);

        return null;
    }

    public void syncTag(NebulaStorageConnectionProvider storageConnectionProvider, NebulaMetaConnectionProvider metaConnectionProvider,
                        NebulaGraphConnectionProvider graphConnectionProvider) throws Exception {
        MetaClient metaClient = metaConnectionProvider.getMetaClient();
        StorageClient storageClient = storageConnectionProvider.getStorageClient();
        List<TagItem> tagItemList = metaClient.getTags("demo_basketballplayer");
        for (TagItem item : tagItemList) {
            List<String> fieldList = new ArrayList<>();
            List<Integer> positionList = new ArrayList<>();
            int index = 1;
            for (ColumnDef columnDef : item.getSchema().getColumns()) {
                fieldList.add(new String(columnDef.getName()));
                positionList.add(index);
                index++;
            }

            VertexExecutionOptions addVertexExecutionOptions = new VertexExecutionOptions.ExecutionOptionBuilder()
                    .setGraphSpace("test")
                    .setTag(new String(item.getTag_name()))
                    .setIdIndex(0)
                    .setFields(fieldList)
                    .setPositions(positionList)
                    .setBatchSize(100)
                    .build();

            NebulaVertexBatchOutputFormat nebulaBatchOutputFormat = new NebulaVertexBatchOutputFormat(graphConnectionProvider, metaConnectionProvider, addVertexExecutionOptions);
            nebulaBatchOutputFormat.open(0, 1);

            ExecutionOptions vertexExecutionOptions = new VertexExecutionOptions.ExecutionOptionBuilder()
                    .setGraphSpace("demo_basketballplayer")
                    .setTag(new String(item.tag_name))
                    .setFields(Arrays.asList())
                    .setLimit(100)
                    .build();

            NebulaVertexSource nebulaSource = new NebulaVertexSource(storageClient, vertexExecutionOptions);

            List<org.apache.flink.types.Row> list = new ArrayList<>();
            while (nebulaSource.hasNext()) {
                BaseTableRow row = nebulaSource.next();
                List<ValueWrapper> values = row.getValues();
                org.apache.flink.types.Row record = new Row(values.size());
                for (int i = 0; i < values.size(); i++) {
                    record.setField(i, getValue(values.get(i)));
                }
                list.add(record);
                nebulaBatchOutputFormat.writeRecord(record);
            }
            nebulaBatchOutputFormat.flush();
            log.info("数据：{}", list);
        }
    }

    public void syncEdge(NebulaStorageConnectionProvider storageConnectionProvider, NebulaMetaConnectionProvider metaConnectionProvider,
                         NebulaGraphConnectionProvider graphConnectionProvider) throws Exception {
        MetaClient metaClient = metaConnectionProvider.getMetaClient();
        StorageClient storageClient = storageConnectionProvider.getStorageClient();
        List<EdgeItem> edgeItemList = metaClient.getEdges("demo_basketballplayer");
        for (EdgeItem item : edgeItemList) {
            List<String> fieldList = new ArrayList<>();
            List<Integer> positionList = new ArrayList<>();
            int index = 3;
            for (ColumnDef columnDef : item.getSchema().getColumns()) {
                fieldList.add(new String(columnDef.getName()));
                positionList.add(index);
                index++;
            }

            EdgeExecutionOptions addEdgeExecutionOptions = new EdgeExecutionOptions.ExecutionOptionBuilder()
                    .setGraphSpace("test")
                    .setEdge(new String(item.getEdge_name()))
                    .setSrcIndex(0)
                    .setDstIndex(1)
                    .setRankIndex(2)
                    .setFields(fieldList)
                    .setPositions(positionList)
                    .setBatchSize(100)
                    .build();

            NebulaEdgeBatchOutputFormat nebulaBatchOutputFormat = new NebulaEdgeBatchOutputFormat(graphConnectionProvider, metaConnectionProvider, addEdgeExecutionOptions);
            nebulaBatchOutputFormat.open(0, 1);

            ExecutionOptions edgeExecutionOptions = new EdgeExecutionOptions.ExecutionOptionBuilder()
                    .setGraphSpace("demo_basketballplayer")
                    .setEdge(new String(item.getEdge_name()))
                    .setFields(Arrays.asList())
                    .setLimit(100)
                    .build();

            NebulaEdgeSource nebulaSource = new NebulaEdgeSource(storageClient, edgeExecutionOptions);

            List<org.apache.flink.types.Row> list = new ArrayList<>();
            while (nebulaSource.hasNext()) {
                BaseTableRow row = nebulaSource.next();
                List<ValueWrapper> values = row.getValues();
                org.apache.flink.types.Row record = new Row(values.size());
                for (int i = 0; i < values.size(); i++) {
                    record.setField(i, getValue(values.get(i)));
                }
                list.add(record);
                nebulaBatchOutputFormat.writeRecord(record);
            }
            nebulaBatchOutputFormat.flush();
            log.info("数据：{}", list);
        }
    }

    public Object getValue(ValueWrapper value) throws UnsupportedEncodingException {
        if (value.isLong()) {
            return value.asLong();
        } else if (value.isBoolean()) {
            return value.asBoolean();
        } else if (value.isDouble()) {
            return value.asDouble();
        } else if (value.isString()) {
            return value.asString();
        } else if (value.isTime()) {
            TimeWrapper timeWrapper = value.asTime();
            return timeWrapper.getLocalTimeStr();
        } else if (value.isDate()) {
            DateWrapper dateWrapper = value.asDate();
            return dateWrapper.toString();
        } else if (value.isDateTime()) {
            DateTimeWrapper dateTimeWrapper = value.asDateTime();
            return dateTimeWrapper.getUTCDateTimeStr();
        } else if (value.isVertex()) {
            return value.asNode();
        } else if (value.isEdge()) {
            return value.asRelationship();
        } else if (value.isPath()) {
            return value.asPath();
        } else if (value.isMap()) {
            return value.asMap();
        } else if (value.isEmpty() || value.isNull()) {
            return "";
        } else {
            return value.getValue();
        }
    }

}
